#include "MQTTClient.h"

MQTTClient::MQTTClient(const string &pubTopic, const string &subTopic, int qos)
    : pubTopic(pubTopic),
      subTopic(subTopic),
      qos(qos)
{
    // 初始化
    int ret = mosquitto_lib_init();
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_lib_init,errmsg:%s\n", mosquitto_strerror(ret));
        return;
    }

    mos = mosquitto_new(NULL, true, this);
    if (!mos)
    {
        fprintf(stderr, "Fail to mosquitto_new,errmsg:%s\n", mosquitto_strerror(ret));
        return;
    }

    // 收到消息回调的函数
    mosquitto_message_callback_set(mos, message);
}

MQTTClient::~MQTTClient()
{
    mosquitto_destroy(mos);
    // 清除
    int ret = mosquitto_lib_cleanup();
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_lib_cleanup,errmsg:%s\n", mosquitto_strerror(ret));
    }
}

bool MQTTClient::connect(const string &ip, uint16_t port)
{
    // 链接
    int ret = mosquitto_connect(mos, ip.c_str(), port, 60);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_connect,errmsg:%s\n", mosquitto_strerror(ret));
        return false;
    }

    // 订阅主题 消息订阅 qos 消息质量
    ret = mosquitto_subscribe(mos, NULL, subTopic.c_str(), qos);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_subscribe,errmsg:%s\n", mosquitto_strerror(ret));
        return false;
    }

    // 网络事件循环处理函数，
    // 通过创建新的线程不断调用mosquitto_loop() 函数处理网络事件，不阻塞
    ret = mosquitto_loop_start(mos);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_loop_start,errmsg:%s\n", mosquitto_strerror(ret));
        return false;
    }

    return true;
}

bool MQTTClient::disconnect(void)
{
    int ret = mosquitto_disconnect(mos);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_disconnect,errmsg:%s\n", mosquitto_strerror(ret));
        return false;
    }

    // 网络事件阻塞回收结束处理函数，这是线程客户端接口的一部分。调用一次可停止先前使用
    ret = mosquitto_loop_stop(mos, true);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_loop_stop,errmsg:%s\n", mosquitto_strerror(ret));
        return false;
    }

    return false;
}

int MQTTClient::send(const void *data, uint32_t len)
{
    int ret = mosquitto_publish(mos, NULL, pubTopic.c_str(), len, data, qos, false);
    if (ret != MOSQ_ERR_SUCCESS)
    {
        fprintf(stderr, "Fail to mosquitto_publish,errmsg:%s\n", mosquitto_strerror(ret));
        return -1;
    }

    return len;
}

// 主线程
int MQTTClient::recv(void *data, uint32_t len)
{
    qmutex.lock();
    // 没有消息就阻塞，有消息就则获取
    if (msgQueue.empty())
    {
        qmutex.unlock();
        wait();
        qmutex.lock();
    }

    MQTTMessage msg = msgQueue.front();
    if (msg.size() > len)
    {
        qmutex.unlock();
        return -1;
    }

    memcpy(data, msg.data(), msg.size());

    msgQueue.pop();
    qmutex.unlock();

    return msg.size();
}

// 收到消息的回调函数 子线程 子现场通知主线程
// 静态成员函数
void MQTTClient::message(mosquitto *, void *obj, const mosquitto_message *msg)
{
    cout << "========消息回调函数打印的================" << endl;
    printf("msg topic:%s\n", msg->topic);
    printf("msg data:%s\n", (char *)msg->payload);

    MQTTClient *client = (MQTTClient *)obj;

    client->qmutex.lock();
    client->msgQueue.push(MQTTMessage(msg));
    client->qmutex.unlock();

    client->notify();
}

void MQTTClient::wait()
{
    unique_lock<std::mutex> lock(mutex);
    cv.wait(lock);
}

void MQTTClient::notify()
{
    cv.notify_one();
}
